﻿using System;
using System.Collections.Generic;
using System.Data.Services.Client;
using System.Linq;
using System.Linq.Expressions;
using System.Net;
using System.Reflection;
using AntsCode.StorageClientExtensions.Helpers;
using Microsoft.WindowsAzure;
using Microsoft.WindowsAzure.StorageClient;
using System.Text;

namespace AntsCode.StorageClientExtensions.Denormalization
{
    internal class DenormalizationProcessor
    {
        private string baseAddress;
        private StorageCredentials credentials;

        public DenormalizationProcessor(string baseAddress, StorageCredentials credentials)
        {
            this.baseAddress = baseAddress;
            this.credentials = credentials;
        }

        public void DenormalizeContext(TableServiceContext sourceContext)
        {
            // Are there any changed entities?
            var changedEntity = (from e in sourceContext.Entities
                                 where e.State != EntityStates.Unchanged
                                 select e).FirstOrDefault();

            if (changedEntity == null)
            {
                // Nothing to process
                return;
            }

            // Since all entities must be of the same type when adding or updating,
            // we retrieve the Denormalize Targets for the first entity to improve efficiency.
            var entityType = changedEntity.Entity.GetType();

            DenormalizationRules rules = new DenormalizationRules(entityType);

            if (this.DenormalizingEntity != null)
            {
                // Fire the event handler to populate the rules for this entity type
                this.DenormalizingEntity(this, new DenormalizingEntityEventArgs(rules));
            }

            if (rules.Count() == 0)
            {
                // Nothing to process
                return;
            }

            // Process each entity
            foreach (EntityDescriptor entityDescriptor in sourceContext.Entities)
            {
                switch (entityDescriptor.State)
                {
                    case EntityStates.Added:
                        // Entity has been added, so reverse-denormalize based on foreign keys
                        ReverseDenormalizeEntity(baseAddress, credentials, entityDescriptor.Entity, rules);

                        // Denormalize to all targets
                        DenormalizeEntity(baseAddress, credentials, entityDescriptor.Entity, rules, false, false, false);
                        break;

                    case EntityStates.Modified:
                        // Entity has been modified, so denormalize to all targets
                        DenormalizeEntity(baseAddress, credentials, entityDescriptor.Entity, rules, false, true, false);
                        break;

                    case EntityStates.Deleted:
                        // Entity has been deleted, so clear any denormalized targets
                        DenormalizeEntity(baseAddress, credentials, entityDescriptor.Entity, rules, true, false, false);
                        break;
                }
            }
        }

        public void Denormalize()
        {
            DenormalizeType(null);
        }

        public void DenormalizeType(Type entityType)
        {
            DenormalizationRules rules = new DenormalizationRules(entityType);

            if (this.DenormalizingEntity != null)
            {
                // Fire the event handler to populate the rules for the provided entity type
                this.DenormalizingEntity(this, new DenormalizingEntityEventArgs(rules));
            }

            if (rules.Count() == 0)
            {
                // Nothing to process
                return;
            }

            // Group the rules by source type for processing
            var groups = rules.GroupBy(r => r.SourceType);

            foreach (var group in groups)
            {
                // Denormalize the entities for this type
                Activator.CreateInstance(typeof(DenormalizeEntities<>)
                    .MakeGenericType(group.Key),
                    new object[] 
                    { 
                        baseAddress,
                        credentials,
                        group,
                        this
                    });
            }
        }

        private static void ReverseDenormalizeEntity(string baseAddress, StorageCredentials credentials, object entity, DenormalizationRules rules)
        {
            foreach (var rule in rules)
            {
                if (rule.DestinationType == entity.GetType())
                {
                    // Reverse-denormalize this foreign key
                    Activator.CreateInstance(typeof(ReverseDenormalizeProperties<,>)
                        .MakeGenericType(rule.SourceType, rule.DestinationType),
                        new object[] 
                        { 
                            baseAddress,
                            credentials,
                            entity,
                            rule,
                        });
                }
            }
        }

        private void DenormalizeEntity(string baseAddress, StorageCredentials credentials, object entity, IEnumerable<DenormalizationRule> rules, bool clear, bool changedOnly, bool forceSyncronous)
        {
            foreach (var rule in rules)
            {
                if (rule.SourceType == entity.GetType())
                {
                    // Update the entities for this rule
                    IDenormalizeProperties denormalizeProperties =
                        (IDenormalizeProperties)Activator.CreateInstance(typeof(DenormalizeProperties<,>)
                        .MakeGenericType(rule.SourceType, rule.DestinationType));

                    if (this.Error != null)
                    {
                        // Attach error event handler to catch exceptions in other threads
                        denormalizeProperties.Error += this.Error;
                    }

                    // Run the denormalization
                    denormalizeProperties.Run(
                        baseAddress,
                        credentials,
                        entity,
                        rule,
                        clear,
                        changedOnly,
                        forceSyncronous);
                }
            }
        }

        private static string GetPropertyName(Expression expression)
        {
            string propertyName = null;
            Type expressionType = expression.GetType();

            if (expressionType == typeof(MemberExpression) || expressionType.IsSubclassOf(typeof(MemberExpression)))
            {
                propertyName = ((MemberExpression)expression).Member.Name;
            }
            else if (expressionType == typeof(UnaryExpression) || expressionType.IsSubclassOf(typeof(MemberExpression)))
            {
                propertyName = ((MemberExpression)((UnaryExpression)expression).Operand).Member.Name;
            }

            return propertyName;
        }

        private static PropertyInfo GetProperty(Expression expression, string propertyName)
        {
            PropertyInfo property = null;
            Type expressionType = expression.GetType();

            if (expressionType == typeof(MemberExpression) || expressionType.IsSubclassOf(typeof(MemberExpression)))
            {
                property = ((MemberExpression)expression).Member.DeclaringType.GetProperty(propertyName);
            }
            else if (expressionType == typeof(UnaryExpression) || expressionType.IsSubclassOf(typeof(MemberExpression)))
            {
                property = ((MemberExpression)((UnaryExpression)expression).Operand).Member.DeclaringType.GetProperty(propertyName);
            }

            return property;
        }

        private static object GetPropertyValue(object obj, string propertyName)
        {
            var property = obj.GetType().GetProperty(propertyName);
            return property.GetValue(obj, null);
        }

        private class DenormalizeEntities<TSource>
            where TSource : TableServiceEntity
        {
            public DenormalizeEntities(
                string baseAddress,
                StorageCredentials credentials,
                IEnumerable<DenormalizationRule> rules,
                DenormalizationProcessor processor)
            {
                var tableName = typeof(TSource).Name;

                var context = new TableServiceContext(baseAddress, credentials)
                {
                    IgnoreResourceNotFoundException = true
                };

                // Build query
                var query = (DataServiceQuery<TSource>)context.CreateQuery<TSource>(tableName);

                // Execute query until there is no further continuation
                DataServiceQuery<TSource> currentQuery;
                QueryOperationResponse<TSource> response;
                string nextPartitionKey = null;
                string nextRowKey = null;

                do
                {
                    // Add any continuation token to the current query
                    currentQuery = query;
                    currentQuery = AddContinuationToken(currentQuery, nextPartitionKey, nextRowKey);

                    response = (QueryOperationResponse<TSource>)currentQuery.Execute();

                    // Denormalize each entity in this response
                    foreach (var entity in response)
                    {
                        processor.DenormalizeEntity(baseAddress, credentials, entity, rules, false, false, true);
                    }

                    // Look for a continuation token in this response
                    GetContinuationToken(response, out nextPartitionKey, out nextRowKey);
                }
                while (nextPartitionKey != null || nextRowKey != null);
            }
        }

        private class ReverseDenormalizeProperties<TSource, TDestination>
            where TSource : TableServiceEntity
            where TDestination : TableServiceEntity
        {
            public ReverseDenormalizeProperties(
                string baseAddress,
                StorageCredentials credentials,
                object entity,
                DenormalizationRule rule)
            {
                var context = new TableServiceContext(baseAddress, credentials)
                {
                    IgnoreResourceNotFoundException = true
                };

                var sourceProperties = rule.SourceProperties as Expression<Func<TSource, object>>[];
                var destProperties = rule.DestinationProperties as Expression<Func<TDestination, object>>[];
                var filter = rule.ReverseFilter as Expression<Func<TSource, TDestination, bool>>;

                if (filter == null)
                {
                    // No reverse filter specified, so return
                    return;
                }

                if (ArePropertyValuesPopulated(entity, destProperties))
                {
                    // All properties already have a value, so don't reverse denormalize
                    return;
                }

                // Load the referenced entity based on the foreign key value
                var tableName = typeof(TSource).Name;

                // Evaluate the filter against the source entity
                LambdaExpression evaluatedFilter;

                try
                {
                    evaluatedFilter = ExpressionExtensions.EvaluateParameter(filter, filter.Parameters[1], entity);
                }
                catch (NullNemberException)
                {
                    // The filter econtains a null member vale
                    // Null values are not supported by table storage so we cannot process this rule
                    // (If the member is null, we assume there are no entities to denormalize to)
                    return;
                }

                var where = Expression.Lambda<Func<TSource, bool>>(evaluatedFilter.Body, filter.Parameters[0]);
                object referencedEntity = context.CreateQuery<TSource>(tableName).Where(where).FirstOrDefault();

                if (referencedEntity != null)
                {
                    string sourcePropertyName;
                    string destPropertyName;

                    for (int i = 0; i < sourceProperties.Length; i++)
                    {
                        sourcePropertyName = GetPropertyName(sourceProperties[i].Body);
                        destPropertyName = GetPropertyName(destProperties[i].Body);

                        // Get the value of the source property
                        var sourceProperty = typeof(TSource).GetProperty(sourcePropertyName);
                        var sourceValue = sourceProperty.GetValue(referencedEntity, null);

                        // Set against destination property of the supplied entity
                        var destProperty = entity.GetType().GetProperty(destPropertyName);

                        try
                        {
                            destProperty.SetValue(entity, sourceValue, null);
                        }
                        catch (Exception e)
                        {
                            throw new DenormalizationException(string.Format(
                                "Error updating property {0} in table {1} to value {2}. PartitionKey: {3}, RowKey: {4}.",
                                destProperty.Name,
                                typeof(TDestination).Name,
                                sourceValue,
                                entity.GetType().GetProperty("PartitionKey").GetValue(entity, null),
                                entity.GetType().GetProperty("RowKey").GetValue(entity, null)), e);
                        }
                    }

                    // Trigger ReverseDenormalized event
                    if (rule.OnReverseDenormalized != null)
                    {
                        var onReverseDenormalized = (DenormalizedDelegate<TSource, TDestination>)rule.OnReverseDenormalized;
                        onReverseDenormalized((TSource)referencedEntity, (TDestination)entity);
                    }
                }
            }

            private bool ArePropertyValuesPopulated(object entity, Expression<Func<TDestination, object>>[] destProperties)
            {
                var propertyValuesPopulated = true;
                string destPropertyName;
                object destPropertyValue;

                foreach (var destProperty in destProperties)
                {
                    destPropertyName = GetPropertyName(destProperty.Body);
                    destPropertyValue = GetPropertyValue(entity, destPropertyName);

                    if (IsValueEmpty(destPropertyValue))
                    {
                        propertyValuesPopulated = false;
                        break;
                    }
                }

                return propertyValuesPopulated;
            }

            private bool IsValueEmpty(object value)
            {
                if (value == null)
                {
                    return true;
                }

                var valueType = value.GetType();
                if (valueType == typeof(Guid) && value.Equals(Guid.Empty))
                {
                    return true;
                }
                else if (valueType == typeof(int) && value.Equals(0))
                {
                    return true;
                }
                else if (valueType == typeof(double) && value.Equals(0.0))
                {
                    return true;
                }
                else if (valueType == typeof(DateTime) && value.Equals(new DateTime()))
                {
                    return true;
                }
                else if (valueType == typeof(bool) && value.Equals(false))
                {
                    return true;
                }

                return false;
            }
        }

        private interface IDenormalizeProperties
        {
            void Run(
                 string baseAddress,
                 StorageCredentials credentials,
                 object entity,
                 DenormalizationRule rule,
                 bool clearValue,
                 bool changedOnly,
                 bool forceSyncronous);

            event EventHandler<DenormalizationErrorEventArgs> Error;
        }

        private class DenormalizeProperties<TSource, TDestination> : IDenormalizeProperties
            where TSource : TableServiceEntity
            where TDestination : TableServiceEntity
        {
            // The maximum batch size supported by Table Storage
            private readonly int maxBatchSize = 100;
            private delegate void RunDelegate(
                string baseAddress,
                StorageCredentials credentials,
                object entity,
                DenormalizationRule rule,
                bool clearValue,
                bool changedOnly);

            public void Run(
                string baseAddress,
                StorageCredentials credentials,
                object entity,
                DenormalizationRule rule,
                bool clearValue,
                bool changedOnly,
                bool forceSyncronous)
            {
                var runDelegate = new RunDelegate(RunInternal);

                if (rule.RunAsync && !forceSyncronous)
                {
                    // Run asyncronously in a new thread
                    runDelegate.BeginInvoke(
                        baseAddress,
                        credentials,
                        entity,
                        rule,
                        clearValue,
                        changedOnly,
                        null, null);
                }
                else
                {
                    // Run syncronously in a current thread
                    runDelegate.Invoke(
                        baseAddress,
                        credentials,
                        entity,
                        rule,
                        clearValue,
                        changedOnly);
                }
            }

            private void RunInternal(
                string baseAddress,
                StorageCredentials credentials,
                object entity,
                DenormalizationRule rule,
                bool clearValue,
                bool changedOnly)
            {
                try
                {
                    var context = new TableServiceContext(baseAddress, credentials)
                    {
                        IgnoreResourceNotFoundException = true
                    };

                    var sourceProperties = rule.SourceProperties as Expression<Func<TSource, object>>[];
                    var destinationProperties = rule.DestinationProperties as Expression<Func<TDestination, object>>[];
                    var filter = rule.Filter as Expression<Func<TSource, TDestination, bool>>;

                    if (!clearValue && changedOnly && !this.HavePropertyValuesChanged(context, entity, sourceProperties))
                    {
                        // Property value has not changed to no need to denormalize
                        return;
                    }

                    // Get the property values to denormlize
                    PropertyInfo[] destProperties = new PropertyInfo[sourceProperties.Length];
                    object[] propertyValues = new object[sourceProperties.Length];
                    string sourcePropertyName;
                    string destPropertyName;

                    for (int i = 0; i < sourceProperties.Length; i++)
                    {
                        sourcePropertyName = GetPropertyName(sourceProperties[i].Body);
                        destPropertyName = GetPropertyName(destinationProperties[i].Body);
                        destProperties[i] = GetProperty(destinationProperties[i].Body, destPropertyName);

                        if (clearValue)
                        {
                            propertyValues[i] = GetClearedValue(entity, sourcePropertyName);
                        }
                        else
                        {

                            propertyValues[i] = GetPropertyValue(entity, sourcePropertyName);
                        }
                    }

                    // Build query
                    var tableName = typeof(TDestination).Name;

                    // Evaluate the filter against the source entity
                    LambdaExpression evaluatedFilter;

                    try
                    {
                        evaluatedFilter = ExpressionExtensions.EvaluateParameter(filter, filter.Parameters[0], entity);
                    }
                    catch (NullNemberException)
                    {
                        // The filter econtains a null member vale
                        // Null values are not supported by table storage so we cannot process this rule
                        // (If the member is null, we assume there are no entities to denormalize to)
                        return;
                    }

                    var where = Expression.Lambda<Func<TDestination, bool>>(evaluatedFilter.Body, filter.Parameters[1]);

                    // Build query
                    var query = (DataServiceQuery<TDestination>)context.CreateQuery<TDestination>(tableName).Where(where);

                    // Execute query until there is no further continuation
                    DataServiceQuery<TDestination> currentQuery;
                    QueryOperationResponse<TDestination> response;
                    string nextPartitionKey = null;
                    string nextRowKey = null;
                    bool success;

                    do
                    {
                        // Add any continuation token to the current query
                        currentQuery = query;
                        currentQuery = AddContinuationToken(currentQuery, nextPartitionKey, nextRowKey);

                        do
                        {
                            try
                            {
                                response = (QueryOperationResponse<TDestination>)currentQuery.Execute();
                            }
                            catch (Exception e)
                            {
                                throw new DenormalizationException(string.Format("Error running query {0} for table {1}.",
                                    currentQuery.ToString(),
                                    tableName), e);
                            }

                            // Update the entities in this batch
                            success = this.UpdateBatch(context, rule, tableName, response, destProperties, propertyValues, entity);

                            // If any update in the batch fails then re-run the query and try again
                            // This will only occur of data was updated since the query was run
                        }
                        while (!success);

                        // Look for a continuation token in this response
                        GetContinuationToken(response, out nextPartitionKey, out nextRowKey);
                    }
                    while (nextPartitionKey != null || nextRowKey != null);
                }
                catch (Exception e)
                {
                    if (this.Error != null)
                    {
                        // Since this operation may be run in a separate thread, we need to pass the exception
                        // back to the calling thread
                        this.Error(this, new DenormalizationErrorEventArgs(e));
                    }
                }
            }

            private bool HavePropertyValuesChanged(TableServiceContext context, object entity, Expression<Func<TSource, object>>[] sourceProperties)
            {
                var propertyValueChanged = false;
                var entityName = entity.GetType().Name;
                var partitionKey = ((TSource)entity).PartitionKey;
                var rowKey = ((TSource)entity).RowKey;
                string sourcePropertyName;

                // Retrieve the entity from table storage
                var prevEntity = (from e in context.CreateQuery<TSource>(entityName)
                                  where e.PartitionKey == partitionKey &&
                                  e.RowKey == rowKey
                                  select e).FirstOrDefault();

                if (prevEntity != null)
                {
                    // Have any of the values changed since last saved?
                    foreach (var sourceProperty in sourceProperties)
                    {
                        sourcePropertyName = GetPropertyName(sourceProperty.Body);
                        var existingValue = GetPropertyValue(prevEntity, sourcePropertyName);
                        var currentValue = GetPropertyValue(entity, sourcePropertyName);

                        if ((existingValue == null && currentValue != null) ||
                            (existingValue != null && currentValue == null) ||
                            (existingValue != null && currentValue != null && !existingValue.Equals(currentValue)))
                        {
                            propertyValueChanged = true;
                            break;
                        }
                    }
                }

                return propertyValueChanged;
            }

            private bool UpdateBatch(TableServiceContext context, DenormalizationRule rule, string tableName, QueryOperationResponse<TDestination> entities,
                PropertyInfo[] updateProperties, object[] values, object sourceEntity)
            {
                var rowKeyProperty = typeof(TDestination).GetProperty("RowKey");
                var partitionKeyProperty = typeof(TDestination).GetProperty("PartitionKey");
                string currentPartition = null;
                string initialRowKey;
                string initialPartitionKey;
                string currentRowKey;
                string currentPartitionKey;
                bool success = true;
                int batchCount = 0;
                var entityList = entities.ToList();

                try
                {
                    foreach (var entity in entityList)
                    {
                        // Snapshot the RowKey and PartitionKey before updating property values
                        initialRowKey = (string)rowKeyProperty.GetValue(entity, null);
                        initialPartitionKey = (string)partitionKeyProperty.GetValue(entity, null);

                        // Update the property values
                        for (int i = 0; i < updateProperties.Length; i++)
                        {
                            try
                            {
                                updateProperties[i].SetValue(entity, values[i], null);
                            }
                            catch (Exception e)
                            {
                                // Could not update property, so provide extra information about the property
                                throw new DenormalizationException(string.Format(
                                    "Error updating property {0} in table {1} to value {2}. PartitionKey: {3}, RowKey: {4}.",
                                    updateProperties[i].Name,
                                    tableName,
                                    values[i],
                                    initialPartitionKey,
                                    initialRowKey), e);
                            }
                        }

                        currentRowKey = (string)rowKeyProperty.GetValue(entity, null);
                        currentPartitionKey = (string)partitionKeyProperty.GetValue(entity, null);

                        // Does this entity belong to a different partition?
                        if (currentPartition != null && batchCount > 0 && currentPartitionKey != currentPartition)
                        {
                            // Yes, so save the current batch before continuing
                            context.SaveChangesWithRetries(SaveChangesOptions.ReplaceOnUpdate);
                            batchCount = 0;
                        }

                        // Has updating the properties resulted in the RowKey or PartitionKey of the entity changing?
                        if (currentRowKey != initialRowKey || currentPartitionKey != initialPartitionKey)
                        {
                            // Yes, so an update operation will fail
                            // Instead, we need to delete then re-add the entity
                            rowKeyProperty.SetValue(entity, initialRowKey, null);
                            partitionKeyProperty.SetValue(entity, initialPartitionKey, null);

                            context.DeleteObject(entity);
                            context.SaveChangesWithRetries(SaveChangesOptions.ReplaceOnUpdate);

                            // Entity deleted successfully, now re-add
                            rowKeyProperty.SetValue(entity, currentRowKey, null);
                            partitionKeyProperty.SetValue(entity, currentPartitionKey, null);

                            context.Detach(entity);
                            context.AddObject(tableName, entity);

                            try
                            {
                                context.SaveChangesWithRetries(SaveChangesOptions.ReplaceOnUpdate);
                            }
                            catch (Exception e)
                            {
                                // There is a chance that an entity with the new row/partition key combo already
                                // exists (among other possible causes). Since we've deleted the original entity,
                                // we cannot risk losing the data so generate a GUID RowKey and try re-saving.
                                string rowGuid = Guid.NewGuid().ToString("N");
                                rowKeyProperty.SetValue(entity, rowGuid, null);
                                context.SaveChangesWithRetries(SaveChangesOptions.ReplaceOnUpdate);

                                // Throw an exception
                                throw new DenormalizationException(string.Format(
                                    "Error re-adding entity after deletion. PartitionKey: {0}, RowKey: {1}. Saved to RowKey: {2}.",
                                    currentPartitionKey, currentRowKey, rowGuid), e);
                            }

                            batchCount = 0;
                        }
                        else
                        {
                            // Keys have not changed, so flag the entity for updating
                            context.UpdateObject(entity);
                        }

                        currentPartition = currentPartitionKey;
                        batchCount++;

                        // Have we reached the batch maximum?
                        if (batchCount == maxBatchSize)
                        {
                            // Yes, so save the current batch
                            context.SaveChangesWithRetries(SaveChangesOptions.ReplaceOnUpdate);
                            batchCount = 0;
                        }

                        // Trigger Denormlized event
                        if (rule.OnDenormalized != null)
                        {
                            var onDenormalized = (DenormalizedDelegate<TSource, TDestination>)rule.OnDenormalized;
                            onDenormalized((TSource)sourceEntity, entity);
                        }
                    }

                    if (batchCount > 0)
                    {
                        // Save any remaining entities
                        context.SaveChangesWithRetries(SaveChangesOptions.ReplaceOnUpdate);
                    }
                }
                catch (DataServiceRequestException e)
                {
                    if (e.InnerException is DataServiceClientException &&
                        ((DataServiceClientException)e.InnerException).StatusCode == (int)HttpStatusCode.PreconditionFailed)
                    {
                        // The conditional header was not met, meaning that the data has been updated
                        // since the query was run, return a failure
                        success = false;
                    }
                    else
                    {
                        // Some other exception occurred, so re-throw
                        throw new DenormalizationException(string.Format(
                            "Error updating entity batch in table {0}.{1}",
                            tableName,
                            GetEntityDebugString(entityList, updateProperties, values)), e);
                    }
                }

                return success;
            }

            private string GetEntityDebugString<T>(IEnumerable<T> entities, PropertyInfo[] updateProperties, object[] values)
            {
                var sb = new StringBuilder();
                var rowKeyProperty = typeof(T).GetProperty("RowKey");
                var partitionKeyProperty = typeof(T).GetProperty("PartitionKey");
                string rowKey, partitionKey;

                sb.AppendLine();
                sb.AppendLine("Property values:");

                for (int i = 0; i < updateProperties.Length; i++)
                {
                    sb.AppendLine(string.Format("Property: {0}, Value: {1}", updateProperties[i].Name, values[i]));
                }

                sb.AppendLine("Entities in batch:");

                foreach (var entity in entities)
                {
                    rowKey = (string)rowKeyProperty.GetValue(entity, null);
                    partitionKey = (string)partitionKeyProperty.GetValue(entity, null);

                    sb.AppendLine(string.Format("PartitionKey: {0}, RowKey: {1}", partitionKey, rowKey));
                }

                return sb.ToString();
            }

            private object GetClearedValue(object obj, string propertyName)
            {
                var property = obj.GetType().GetProperty(propertyName);
                var propertyType = property.PropertyType;
                object clearedValue;

                // Is the property a nullable type?
                if (propertyType.IsGenericType && propertyType.GetGenericTypeDefinition() == typeof(Nullable<>))
                {
                    // Yes, so return null
                    clearedValue = null;
                }
                else
                {
                    // Determine cleared value based on type
                    if (propertyType == typeof(Guid))
                    {
                        clearedValue = Guid.Empty;
                    }
                    else if (propertyType == typeof(int))
                    {
                        clearedValue = 0;
                    }
                    else if (propertyType == typeof(double))
                    {
                        clearedValue = 0.0;
                    }
                    else if (propertyType == typeof(DateTime))
                    {
                        clearedValue = new DateTime();
                    }
                    else if (propertyType == typeof(bool))
                    {
                        clearedValue = false;
                    }
                    else
                    {
                        // String
                        clearedValue = null;
                    }
                }

                return clearedValue;
            }

            public event EventHandler<DenormalizationErrorEventArgs> Error;
        }

        private static void GetContinuationToken<T>(QueryOperationResponse<T> response,
            out string nextPartitionKey, out string nextRowKey)
        {
            nextPartitionKey = null;
            nextRowKey = null;

            response.Headers.TryGetValue("x-ms-continuation-NextPartitionKey", out nextPartitionKey);
            response.Headers.TryGetValue("x-ms-continuation-NextRowKey", out nextRowKey);
        }

        private static DataServiceQuery<T> AddContinuationToken<T>(DataServiceQuery<T> data,
            string nextPartitionKey, string nextRowKey)
        {
            if (nextPartitionKey != null)
            {
                data = data.AddQueryOption("NextPartitionKey", nextPartitionKey);
            }

            if (nextRowKey != null)
            {
                data = data.AddQueryOption("NextRowKey", nextRowKey);
            }

            return data;
        }

        public event EventHandler<DenormalizingEntityEventArgs> DenormalizingEntity;
        public event EventHandler<DenormalizationErrorEventArgs> Error;
    }
}